<!-- build time:Tue Dec 17 2019 19:26:41 GMT+0800 (GMT+08:00) --><!doctype html><html class="theme-next mist" lang="zh-Hans"><head><meta name="generator" content="Hexo 3.8.0"><meta name="google-site-verification" content="7Tau9WyVgxnsEY9oYedu9g0U6_8akOX3wiKbaYcrg9A"><meta name="baidu-site-verification" content="EVwLiaxdxX"><link href="/css/xps13.css" rel="stylesheet" type="text/css"><link href="/css/message.css" rel="stylesheet" type="text/css"><script type="text/javascript" src="/js/jquery-1.11.3.min.js"></script><meta charset="UTF-8"><meta name="viewport" content="width=device-width,initial-scale=1,maximum-scale=1"><meta http-equiv="Cache-Control" content="no-transform"><meta http-equiv="Cache-Control" content="no-siteapp"><link href="/lib/font-awesome/css/font-awesome.min.css?v=4.6.2" rel="stylesheet" type="text/css"><link href="/css/main.css?v=5.1.1" rel="stylesheet" type="text/css"><meta name="keywords" content="Spring Boot,Kafka,"><link rel="alternate" href="/atom.xml" title="MrBird" type="application/atom+xml"><link rel="shortcut icon" type="image/x-icon" href="/favicon.ico?v=5.1.1"><meta name="description" content="Kafka是一个分布式的、可分区的、可复制的消息系统，下面是Kafka的几个基本术语：Kafka将消息以topic为单位进行归纳；将向Kafka topic发布消息的程序成为producers；将预订topics并消费消息的程序成为consumer；Kafka以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个broker。"><meta name="keywords" content="Spring Boot,Kafka"><meta property="og:type" content="article"><meta property="og:title" content="Spring Boot整合Kafka"><meta property="og:url" content="http://mrbird.cc/Spring-Boot-Kafka.html"><meta property="og:site_name" content="MrBird"><meta property="og:description" content="Kafka是一个分布式的、可分区的、可复制的消息系统，下面是Kafka的几个基本术语：Kafka将消息以topic为单位进行归纳；将向Kafka topic发布消息的程序成为producers；将预订topics并消费消息的程序成为consumer；Kafka以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个broker。"><meta property="og:locale" content="zh-Hans"><meta property="og:image" content="http://mrbird.cc/img/140721072031172.png"><meta property="og:image" content="http://mrbird.cc/img/log_anatomy.png"><meta property="og:image" content="http://mrbird.cc/img/log_consumer.png"><meta property="og:image" content="http://mrbird.cc/img/consumer-groups.png"><meta property="og:image" content="http://mrbird.cc/img/QQ截图20190326103520.png"><meta property="og:image" content="http://mrbird.cc/img/QQ截图20190326110506.png"><meta property="og:image" content="http://mrbird.cc/img/QQ截图20190326111559.png"><meta property="og:image" content="http://mrbird.cc/img/QQ截图20190326111928.png"><meta property="og:image" content="http://mrbird.cc/img/QQ截图20190326113911.png"><meta property="og:image" content="http://mrbird.cc/img/QQ截图20190326113950.png"><meta property="og:image" content="http://mrbird.cc/img/QQ截图20190326154735.png"><meta property="og:image" content="http://mrbird.cc/img/QQ截图20190326155948.png"><meta property="og:image" content="http://mrbird.cc/img/QQ图片20190326162014.png"><meta property="og:image" content="http://mrbird.cc/img/QQ截图20190326163502.png"><meta property="og:image" content="http://mrbird.cc/img/QQ截图20190326171125.png"><meta property="og:updated_time" content="2019-10-28T12:14:46.272Z"><meta name="twitter:card" content="summary"><meta name="twitter:title" content="Spring Boot整合Kafka"><meta name="twitter:description" content="Kafka是一个分布式的、可分区的、可复制的消息系统，下面是Kafka的几个基本术语：Kafka将消息以topic为单位进行归纳；将向Kafka topic发布消息的程序成为producers；将预订topics并消费消息的程序成为consumer；Kafka以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个broker。"><meta name="twitter:image" content="http://mrbird.cc/img/140721072031172.png"><script type="text/javascript" id="hexo.configurations">var NexT=window.NexT||{},CONFIG={root:"/",scheme:"Mist",sidebar:{position:"left",display:"always",offset:12,offset_float:0,b2t:!1,scrollpercent:!1},fancybox:!1,motion:!1}</script><title>Spring Boot整合Kafka | MrBird</title></head><body ondragstart="return!1" class="animsition" lang="zh-Hans" style="overflow-x:hidden;padding-right:280px"><script type="text/javascript" src="/js/mo.min.js"></script><style>@media (min-width:768px) and (max-width:991px){body .header-innerr{width:700px!important}}.header-innerr{margin:0 auto;padding:100px 0 70px;width:880px}@media (min-width:1600px){.container .header-innerr{width:1200px}}.header-innerr{position:relative}.header-innerr{padding:20px 0 0}.header-innerr:after,.header-innerr:before{content:" ";display:table}.header-innerr:after{clear:both}@media (max-width:767px){.header-innerr{width:auto;padding:10px;margin-bottom:-20px}}</style><div class="container sidebar-position-left page-post-detail"><div class="headband"></div><header id="header" class="header"><div class="header-inner"><div class="site-brand-wrapper"><div class="site-meta"><link href="https://fonts.font.im/css?family=Merienda" rel="stylesheet"><div class="custom-logo-site-title"><a href="/" class="brand" rel="start"><span class="logo-line-before"><i></i></span> <span class="site-title" style="font-family:Merienda;font-size:1.3rem">MrBird</span> <span class="logo-line-after"><i></i></span></a></div><p class="site-subtitle"></p></div><div class="site-nav-toggle"><button><span class="btn-bar"></span> <span class="btn-bar"></span> <span class="btn-bar"></span></button></div></div><nav class="site-nav"><ul id="menu" class="menu"><li class="menu-item menu-item-home"><a href="/" rel="section">HOME</a></li><li class="menu-item menu-item-archives"><a href="/archives/" rel="section">ARCHIVES</a></li><li class="menu-item menu-item-tags"><a href="/tags/" rel="section">TAGS</a></li><li class="menu-item menu-item-friends"><a href="/friends/" rel="section">FRIENDS</a></li><div class="sidebar-toggle" style="display:none"><div class="sidebar-toggle-line-wrap"><span class="sidebar-toggle-line sidebar-toggle-line-first"></span> <span class="sidebar-toggle-line sidebar-toggle-line-middle"></span> <span class="sidebar-toggle-line sidebar-toggle-line-last"></span></div></div></ul><div class="site-search"><div class="popup search-popup local-search-popup"><div class="local-search-header clearfix"><span class="search-icon"><i class="fa fa-search"></i> </span><span class="popup-btn-close"><i class="fa fa-times-circle"></i></span><div class="local-search-input-wrapper"><input autocomplete="off" placeholder="Search" spellcheck="false" type="text" id="local-search-input"></div></div><div id="local-search-result"></div></div></div></nav></div><div class="header-innerr"><div class="note info" style="margin:0;letter-spacing:.15px">🐤手把手教你搭建<strong>Spring Cloud微服务权限系统</strong>（从零到部署）：<a style="color:#40dab2;font-weight:600" href="https://www.kancloud.cn/mrbird/spring-cloud" target="_blank">https://www.kancloud.cn/mrbird/spring-cloud</a></div></div></header><main id="main" class="main"><div class="main-inner"><div class="content-wrap"><div id="content" class="content"><div id="posts" class="posts-expand"><article class="post post-type-normal" itemscope itemtype="http://schema.org/Article"><link itemprop="mainEntityOfPage" href="http://mrbird.cc/Spring-Boot-Kafka.html"><span hidden itemprop="author" itemscope itemtype="http://schema.org/Person"><meta itemprop="name" content="MrBird"><meta itemprop="description" content=""><meta itemprop="image" content="/images/blogImage.jpg"></span><span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization"><meta itemprop="name" content="MrBird"></span><header class="post-header"><h1 class="post-title" itemprop="name headline">Spring Boot整合Kafka</h1><div class="post-meta"><span class="post-time"><span class="post-meta-item-icon"><i class="fa fa-calendar-o"></i> </span><span class="post-meta-item-text">Posted on</span> <time title="创建于" itemprop="dateCreated datePublished" datetime="2018-11-21T16:26:17+08:00">2018-11-21 </time></span><span></span> <span class="post-meta-divider">|</span> <span class="page-pv"><i class="fa fa-laptop"></i>&nbsp;&nbsp;Visit count <span class="busuanzi-value" id="busuanzi_value_page_pv"></span></span></div></header><div class="post-body" itemprop="articleBody"><p>Kafka是一个分布式的、可分区的、可复制的消息系统，下面是Kafka的几个基本术语：</p><ol><li><p>Kafka将消息以<strong>topic</strong>为单位进行归纳；</p></li><li><p>将向Kafka topic发布消息的程序成为<strong>producers</strong>；</p></li><li><p>将预订topics并消费消息的程序成为<strong>consumer</strong>；</p></li><li><p>Kafka以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个<strong>broker</strong>。</p></li></ol><a id="more"></a><p>producers通过网络将消息发送到Kafka集群，集群向消费者提供消息，如下图所示：</p><p><img src="img/140721072031172.png" alt="140721072031172.png"></p><p>创建一个topic时，可以指定<strong>partitions</strong>（分区）数目，partitions数越多，其吞吐量也越大，但是需要的资源也越多，同时也会导致更高的不可用性，kafka在接收到producers发送的消息之后，会根据均衡策略将消息存储到不同的partitions中：</p><p><img src="img/log_anatomy.png" alt="log_anatomy.png"></p><p>在每个partitions中，消息以顺序存储，最晚接收的的消息会最后被消费。</p><p>producers在向kafka集群发送消息的时候，可以通过指定partitions来发送到指定的partitions中。也可以通过指定均衡策略来将消息发送到不同的partitions中。如果不指定，就会采用默认的随机均衡策略，将消息随机的存储到不同的partitions中。</p><p>在consumer消费消息时，kafka使用offset来记录当前消费的位置：</p><p><img src="img/log_consumer.png" style="width:25rem"></p><p>在kafka的设计中，可以有多个不同的group来同时消费同一个topic下的消息，如图，我们有两个不同的group同时消费，他们的的消费的记录位置offset各不项目，不互相干扰。</p><p>对于一个group而言，consumer的数量不应该多于partitions的数量，因为在一个group中，每个partitions至多只能绑定到一个consumer上，即一个consumer可以消费多个partitions，一个partitions只能给一个consumer消费。因此，若一个group中的consumer数量大于partitions数量的话，多余的consumer将不会收到任何消息。</p><p><img src="img/consumer-groups.png" style="width:25rem"></p><h2 id="Kafka安装使用"><a href="#Kafka安装使用" class="headerlink" title="Kafka安装使用"></a>Kafka安装使用</h2><p>这里演示在Windows下Kafka安装与使用。Kafka下载地址：<a href="http://kafka.apache.org/downloads" target="_blank" rel="noopener">http://kafka.apache.org/downloads</a>，选择二进制文件下载（Binary downloads），然后解压即可。</p><p>Kafka的配置文件位于config目录下，因为Kafka集成了Zookeeper（Kafka存储消息的地方），所以config目录下除了有Kafka的配置文件server.properties外，还可以看到一个Zookeeper配置文件zookeeper.properties：</p><p><img src="img/QQ截图20190326103520.png" alt="QQ截图20190326103520.png"></p><p>打开server.properties，将<code>broker.id</code>的值修改为1，每个broker的id都必须设置为Integer类型，且不能重复。Zookeeper的配置保持默认即可。</p><p>接下来开始使用Kafka。</p><h3 id="启动Zookeeper"><a href="#启动Zookeeper" class="headerlink" title="启动Zookeeper"></a>启动Zookeeper</h3><p>在Windows下执行下面这些命令可能会出现<span style="color:red">找不到或无法加载主类</span>的问题，解决方案可参考：<a href="https://blog.csdn.net/cx2932350/article/details/78870135" target="_blank" rel="noopener">https://blog.csdn.net/cx2932350/article/details/78870135</a>。</p><p>在Kafka根目录下使用cmd执行下面这条命令，启动ZK：</p><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">bin\windows\zookeeper-server-start.bat config\zookeeper.properties</span><br></pre></td></tr></table></figure><p></p><p>在Linux下，可以使用后台进程的方式启动ZK：</p><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">bin/zookeeper-server-start.sh -daemon config/zookeeper.properties</span><br></pre></td></tr></table></figure><p></p><h3 id="启动Kafka"><a href="#启动Kafka" class="headerlink" title="启动Kafka"></a>启动Kafka</h3><p>执行下面这条命令启动Kafka：</p><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">bin\windows\kafka-server-start.bat config\server.properties</span><br></pre></td></tr></table></figure><p></p><p>Linux对应命令：</p><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">bin/kafka-server-start.sh config/server.properties</span><br></pre></td></tr></table></figure><p></p><p>当看到命令行打印如下信息，说明启动完毕:</p><p><img src="img/QQ截图20190326110506.png" alt="QQ截图20190326110506.png"></p><h3 id="创建Topic"><a href="#创建Topic" class="headerlink" title="创建Topic"></a>创建Topic</h3><p>执行下面这条命令创建一个Topic</p><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic <span class="built_in">test</span></span><br></pre></td></tr></table></figure><p></p><p>这条命令的意思是，创建一个Topic到ZK（指定ZK的地址），副本个数为1，分区数为1，Topic的名称为test。</p><p>Linux对应的命令为:</p><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic <span class="built_in">test</span></span><br></pre></td></tr></table></figure><p></p><p>创建好后我们可以查看Kafka里的Topic列表：</p><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">bin\windows\kafka-topics.bat --list --zookeeper localhost:2181</span><br></pre></td></tr></table></figure><p></p><p><img src="img/QQ截图20190326111559.png" alt="QQ截图20190326111559.png"></p><p>可看到目前只包含一个我们刚创建的test Topic。</p><p>Linux对应的命令为：</p><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">bin/kafka-topics.sh --list --zookeeper localhost:2181</span><br></pre></td></tr></table></figure><p></p><p>查看test Topic的具体信息：</p><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">bin\windows\kafka-topics.bat --describe --zookeeper localhost:2181 --topic <span class="built_in">test</span></span><br></pre></td></tr></table></figure><p></p><p><img src="img/QQ截图20190326111928.png" alt="QQ截图20190326111928.png"></p><p>Linux对应的命令为：</p><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic <span class="built_in">test</span></span><br></pre></td></tr></table></figure><p></p><h3 id="生产消息和消费消息"><a href="#生产消息和消费消息" class="headerlink" title="生产消息和消费消息"></a>生产消息和消费消息</h3><p><strong>启动Producers</strong></p><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic <span class="built_in">test</span></span><br></pre></td></tr></table></figure><p></p><p>9092为生产者的默认端口号。这里启动了生产者，准备往test Topic里发送数据。</p><p>Linux下对应的命令为：</p><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">bin/kafka-console-producer.sh --broker-list localhost:9092 --topic <span class="built_in">test</span></span><br></pre></td></tr></table></figure><p></p><p><strong>启动Consumers</strong></p><p>接着启动一个消费者用于消费生产者生产的数据，新建一个cmd窗口，输入下面这条命令：</p><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic <span class="built_in">test</span> --from-beginning</span><br></pre></td></tr></table></figure><p></p><p><code>from-beginning</code>表示从头开始读取数据。</p><p>Linux下对应的命令为：</p><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic <span class="built_in">test</span> --from-beginning</span><br></pre></td></tr></table></figure><p></p><p>启动好生产者和消费者后我们在生产者里生产几条数据:</p><p><img src="img/QQ截图20190326113911.png" alt="QQ截图20190326113911.png"></p><p>消费者成功接收到数据：</p><p><img src="img/QQ截图20190326113950.png" alt="QQ截图20190326113950.png"></p><h2 id="Spring-Boot整合Kafaka"><a href="#Spring-Boot整合Kafaka" class="headerlink" title="Spring Boot整合Kafaka"></a>Spring Boot整合Kafaka</h2><p>上面简单介绍了Kafka的使用，下面我们开始在Spring Boot里使用Kafka。</p><p>新建一个Spring Boot项目，版本为2.1.3.RELEASE，并引入如下依赖:</p><figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br></pre></td><td class="code"><pre><span class="line"> <span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.springframework.boot<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>spring-boot-starter-web<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.springframework.kafka<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>spring-kafka<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br></pre></td></tr></table></figure><p></p><h3 id="生产者配置"><a href="#生产者配置" class="headerlink" title="生产者配置"></a>生产者配置</h3><p>新建一个Java配置类<code>KafkaProducerConfig</code>，用于配置生产者：</p><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Configuration</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">KafkaProducerConfig</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;spring.kafka.bootstrap-servers&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String bootstrapServers;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Bean</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> ProducerFactory&lt;String, String&gt; <span class="title">producerFactory</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        Map&lt;String, Object&gt; configProps = <span class="keyword">new</span> HashMap&lt;&gt;();</span><br><span class="line">        configProps.put(</span><br><span class="line">                ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,</span><br><span class="line">                bootstrapServers);</span><br><span class="line">        configProps.put(</span><br><span class="line">                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,</span><br><span class="line">                StringSerializer.class);</span><br><span class="line">        configProps.put(</span><br><span class="line">                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,</span><br><span class="line">                StringSerializer.class);</span><br><span class="line">        <span class="keyword">return</span> <span class="keyword">new</span> DefaultKafkaProducerFactory&lt;&gt;(configProps);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Bean</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> KafkaTemplate&lt;String, String&gt; <span class="title">kafkaTemplate</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="keyword">return</span> <span class="keyword">new</span> KafkaTemplate&lt;&gt;(producerFactory());</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure><p></p><p>首先我们配置了一个<code>producerFactory</code>，方法里配置了Kafka Producer实例的策略。<code>bootstrapServers</code>为Kafka生产者的地址，我们在配置文件application.yml里配置它：</p><figure class="highlight yml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line"><span class="attr">spring:</span></span><br><span class="line"><span class="attr">  kafka:</span></span><br><span class="line"><span class="attr">    bootstrap-servers:</span> <span class="attr">localhost:9092</span></span><br></pre></td></tr></table></figure><p></p><p><code>ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG</code>和<code>ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG</code>指定了key，value序列化策略，这里指定为Kafka提供的<code>StringSerializer</code>，因为我们暂时只发送简单的String类型的消息。</p><p>接着我们使用<code>producerFactory</code>配置了<code>kafkaTemplate</code>，其包含了发送消息的便捷方法，后面我们就用这个对象来发送消息。</p><h3 id="发布消息"><a href="#发布消息" class="headerlink" title="发布消息"></a>发布消息</h3><p>配置好生产者，我们就可以开始发布消息了。</p><p>新建一个<code>SendMessageController</code>：</p><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@RestController</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">SendMessageController</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Autowired</span></span><br><span class="line">    <span class="keyword">private</span> KafkaTemplate&lt;String, String&gt; kafkaTemplate;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@GetMapping</span>(<span class="string">"send/&#123;message&#125;"</span>)</span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">send</span><span class="params">(@PathVariable String message)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">this</span>.kafkaTemplate.send(<span class="string">"test"</span>, message);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure><p></p><p>我们注入了<code>kafkaTemplate</code>对象，key-value都为String类型，并通过它的<code>send</code>方法来发送消息。其中<code>test</code>为Topic的名称，上面我们已经使用命令创建过这个Topic了。</p><p><code>send</code>方法是一个异步方法，我们可以通过回调的方式来确定消息是否发送成功，我们改造<code>SendMessageController</code>：</p><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@RestController</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">SendMessageController</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">private</span> Logger logger = LoggerFactory.getLogger(<span class="keyword">this</span>.getClass());</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Autowired</span></span><br><span class="line">    <span class="keyword">private</span> KafkaTemplate&lt;String, String&gt; kafkaTemplate;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@GetMapping</span>(<span class="string">"send/&#123;message&#125;"</span>)</span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">send</span><span class="params">(@PathVariable String message)</span> </span>&#123;</span><br><span class="line">        ListenableFuture&lt;SendResult&lt;String, String&gt;&gt; future = <span class="keyword">this</span>.kafkaTemplate.send(<span class="string">"test"</span>, message);</span><br><span class="line">        future.addCallback(<span class="keyword">new</span> ListenableFutureCallback&lt;SendResult&lt;String, String&gt;&gt;() &#123;</span><br><span class="line">            <span class="meta">@Override</span></span><br><span class="line">            <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onSuccess</span><span class="params">(SendResult&lt;String, String&gt; result)</span> </span>&#123;</span><br><span class="line">                logger.info(<span class="string">"成功发送消息：&#123;&#125;，offset=[&#123;&#125;]"</span>, message, result.getRecordMetadata().offset());</span><br><span class="line">            &#125;</span><br><span class="line"></span><br><span class="line">            <span class="meta">@Override</span></span><br><span class="line">            <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onFailure</span><span class="params">(Throwable ex)</span> </span>&#123;</span><br><span class="line">                logger.error(<span class="string">"消息：&#123;&#125; 发送失败，原因：&#123;&#125;"</span>, message, ex.getMessage());</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure><p></p><p>消息发送成功后，会回调<code>onSuccess</code>方法，发送失败后回调<code>onFailure</code>方法。</p><h3 id="消费者配置"><a href="#消费者配置" class="headerlink" title="消费者配置"></a>消费者配置</h3><p>接着我们来配置消费者，新建一个Java配置类<code>KafkaConsumerConfig</code>：</p><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@EnableKafka</span></span><br><span class="line"><span class="meta">@Configuration</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">KafkaConsumerConfig</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;spring.kafka.bootstrap-servers&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String bootstrapServers;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;spring.kafka.consumer.group-id&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String consumerGroupId;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;spring.kafka.consumer.auto-offset-reset&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String autoOffsetReset;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Bean</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> ConsumerFactory&lt;String, String&gt; <span class="title">consumerFactory</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        Map&lt;String, Object&gt; props = <span class="keyword">new</span> HashMap&lt;&gt;();</span><br><span class="line">        props.put(</span><br><span class="line">                ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,</span><br><span class="line">                bootstrapServers);</span><br><span class="line">        props.put(</span><br><span class="line">                ConsumerConfig.GROUP_ID_CONFIG,</span><br><span class="line">                consumerGroupId);</span><br><span class="line">        props.put(</span><br><span class="line">                ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,</span><br><span class="line">                autoOffsetReset);</span><br><span class="line">        props.put(</span><br><span class="line">                ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,</span><br><span class="line">                StringDeserializer.class);</span><br><span class="line">        props.put(</span><br><span class="line">                ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,</span><br><span class="line">                StringDeserializer.class);</span><br><span class="line">        <span class="keyword">return</span> <span class="keyword">new</span> DefaultKafkaConsumerFactory&lt;&gt;(props);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Bean</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; <span class="title">kafkaListenerContainerFactory</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory</span><br><span class="line">                = <span class="keyword">new</span> ConcurrentKafkaListenerContainerFactory&lt;&gt;();</span><br><span class="line">        factory.setConsumerFactory(consumerFactory());</span><br><span class="line">        <span class="keyword">return</span> factory;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure><p></p><p><code>consumerGroupId</code>和<code>autoOffsetReset</code>需要在application.yml里配置：</p><figure class="highlight yml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line"><span class="attr">spring:</span></span><br><span class="line"><span class="attr">  kafka:</span></span><br><span class="line"><span class="attr">    consumer:</span></span><br><span class="line"><span class="attr">      group-id:</span> <span class="string">test-consumer</span></span><br><span class="line"><span class="attr">      auto-offset-reset:</span> <span class="string">latest</span></span><br></pre></td></tr></table></figure><p></p><p>其中<code>group-id</code>将消费者进行分组（你也可以不进行分组），组名为<code>test-consumer</code>，并指定了消息读取策略，包含四个可选值：</p><p><img src="img/QQ截图20190326154735.png" alt="QQ截图20190326154735.png"></p><ul><li><p>earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费</p></li><li><p>latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据</p></li><li><p>none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常</p></li><li><p>exception:直接抛出异常</p></li></ul><p>在<code>KafkaConsumerConfig</code>中我们配置了<code>ConsumerFactory</code>和<code>KafkaListenerContainerFactory</code>。当这两个Bean成功注册到Spring IOC容器中后，我们便可以使用<code>@KafkaListener</code>注解来监听消息了。</p><p>配置类上需要<code>@EnableKafka</code>注释才能在Spring托管Bean上检测<code>@KafkaListener</code>注解。</p><h3 id="消息消费"><a href="#消息消费" class="headerlink" title="消息消费"></a>消息消费</h3><p>配置好消费者，我们就可以开始消费消息了，新建<code>KafkaMessageListener</code>：</p><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Component</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">KafkaMessageListener</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">private</span> Logger logger = LoggerFactory.getLogger(<span class="keyword">this</span>.getClass());</span><br><span class="line"></span><br><span class="line">    <span class="meta">@KafkaListener</span>(topics = <span class="string">"test"</span>, groupId = <span class="string">"test-consumer"</span>)</span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">listen</span><span class="params">(String message)</span> </span>&#123;</span><br><span class="line">        logger.info(<span class="string">"接收消息: &#123;&#125;"</span>, message);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure><p></p><p>我们通过<code>@KafkaListener</code>注解来监听名称为test的Topic，消费者分组的组名为<code>test-consumer</code>。</p><h3 id="演示"><a href="#演示" class="headerlink" title="演示"></a>演示</h3><p>启动Spring Boot项目，启动过程中，控制台会输出Kafka的配置，启动好后，访问<a href="http://localhost:8080/send/hello,mrbird" target="_blank" rel="noopener">http://localhost:8080/send/hello,mrbird</a>，控制台输出如下：</p><p><img src="img/QQ截图20190326155948.png" alt="QQ截图20190326155948.png"></p><h3 id="KafkaListener详解"><a href="#KafkaListener详解" class="headerlink" title="@KafkaListener详解"></a>@KafkaListener详解</h3><p><code>@KafkaListener</code>除了可以指定Topic名称和分组id外，我们还可以同时监听来自多个Topic的消息:</p><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@KafkaListener</span>(topics = <span class="string">"topic1, topic2"</span>)</span><br></pre></td></tr></table></figure><p></p><p>我们还可以通过<code>@Header</code>注解来获取当前消息来自哪个分区（partitions）:</p><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@KafkaListener</span>(topics = <span class="string">"test"</span>, groupId = <span class="string">"test-consumer"</span>)</span><br><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">listen</span><span class="params">(@Payload String message,</span></span></span><br><span class="line"><span class="function"><span class="params">                   @Header(KafkaHeaders.RECEIVED_PARTITION_ID)</span> <span class="keyword">int</span> partition) </span>&#123;</span><br><span class="line">    logger.info(<span class="string">"接收消息: &#123;&#125;，partition：&#123;&#125;"</span>, message, partition);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure><p></p><p>重启项目，再次访问<a href="http://localhost:8080/send/hello,mrbird" target="_blank" rel="noopener">http://localhost:8080/send/hello,mrbird</a>，控制台输出如下：</p><p><img src="img/QQ图片20190326162014.png" alt="QQ图片20190326162014.png"></p><p>因为我们没有进行分区，所以test Topic只有一个区，下标为0。</p><p>我们可以通过<code>@KafkaListener</code>来指定只接收来自特定分区的消息：</p><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@KafkaListener</span>(groupId = <span class="string">"test-consumer"</span>,</span><br><span class="line">        topicPartitions = <span class="meta">@TopicPartition</span>(topic = <span class="string">"test"</span>,</span><br><span class="line">                partitionOffsets = &#123;</span><br><span class="line">                        <span class="meta">@PartitionOffset</span>(partition = <span class="string">"0"</span>, initialOffset = <span class="string">"0"</span>)</span><br><span class="line">            &#125;))</span><br><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">listen</span><span class="params">(@Payload String message,</span></span></span><br><span class="line"><span class="function"><span class="params">                   @Header(KafkaHeaders.RECEIVED_PARTITION_ID)</span> <span class="keyword">int</span> partition) </span>&#123;</span><br><span class="line">    logger.info(<span class="string">"接收消息: &#123;&#125;，partition：&#123;&#125;"</span>, message, partition);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure><p></p><p>如果不需要指定<code>initialOffset</code>，上面代码可以简化为：</p><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@KafkaListener</span>(groupId = <span class="string">"test-consumer"</span>, </span><br><span class="line">	topicPartitions = <span class="meta">@TopicPartition</span>(topic = <span class="string">"test"</span>, partitions = &#123; <span class="string">"0"</span>, <span class="string">"1"</span> &#125;))</span><br></pre></td></tr></table></figure><h3 id="消息过滤器"><a href="#消息过滤器" class="headerlink" title="消息过滤器"></a>消息过滤器</h3><p>我们可以为消息监听添加过滤器来过滤一些特定的信息。我们在消费者配置类<code>KafkaConsumerConfig</code>的<code>kafkaListenerContainerFactory</code>方法里配置过滤规则：</p><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Bean</span></span><br><span class="line"><span class="function"><span class="keyword">public</span> ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; <span class="title">kafkaListenerContainerFactory</span><span class="params">()</span> </span>&#123;</span><br><span class="line">    ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory</span><br><span class="line">            = <span class="keyword">new</span> ConcurrentKafkaListenerContainerFactory&lt;&gt;();</span><br><span class="line">    factory.setConsumerFactory(consumerFactory());</span><br><span class="line">    <span class="comment">// ------- 过滤配置 --------</span></span><br><span class="line">    factory.setRecordFilterStrategy(</span><br><span class="line">            r -&gt; r.value().contains(<span class="string">"fuck"</span>)</span><br><span class="line">    );</span><br><span class="line">    <span class="keyword">return</span> factory;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure><p></p><p><code>setRecordFilterStrategy</code>接收<code>RecordFilterStrategy&lt;K, V&gt;</code>，他是一个函数式接口：</p><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">interface</span> <span class="title">RecordFilterStrategy</span>&lt;<span class="title">K</span>, <span class="title">V</span>&gt; </span>&#123;</span><br><span class="line">    <span class="function"><span class="keyword">boolean</span> <span class="title">filter</span><span class="params">(ConsumerRecord&lt;K, V&gt; var1)</span></span>;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure><p></p><p>所以我们用lambda表达式指定了上面这条规则，即如果消息内容包含<code>fuck</code>这个粗鄙之语的时候，则不接受消息。</p><p>配置好后我们重启项目，分别发送下面这两条请求：</p><ol><li><p><a href="http://localhost:8080/send/fuck,mrbird" target="_blank" rel="noopener">http://localhost:8080/send/fuck,mrbird</a></p></li><li><p><a href="http://localhost:8080/send/love,mrbird" target="_blank" rel="noopener">http://localhost:8080/send/love,mrbird</a></p></li></ol><p>观察控制台：</p><p><img src="img/QQ截图20190326163502.png" alt="QQ截图20190326163502.png"></p><p>可以看到，fuck,mrbird这条消息没有被接收。</p><h3 id="发送复杂的消息"><a href="#发送复杂的消息" class="headerlink" title="发送复杂的消息"></a>发送复杂的消息</h3><p>截至目前位置我们只发送了简单的字符串类型的消息，我们可以自定义消息转换器来发送复杂的消息。</p><p><strong>定义消息实体</strong></p><p>创建一个Message类：</p><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">Message</span> <span class="keyword">implements</span> <span class="title">Serializable</span> </span>&#123;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> <span class="keyword">long</span> serialVersionUID = <span class="number">6678420965611108427L</span>;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">private</span> String from;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">private</span> String message;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="title">Message</span><span class="params">()</span> </span>&#123;</span><br><span class="line">    	</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="title">Message</span><span class="params">(String from, String message)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">this</span>.from = from;</span><br><span class="line">        <span class="keyword">this</span>.message = message;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> String <span class="title">toString</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="keyword">return</span> <span class="string">"Message&#123;"</span> +</span><br><span class="line">                <span class="string">"from='"</span> + from + <span class="string">'\''</span> +</span><br><span class="line">                <span class="string">", message='"</span> + message + <span class="string">'\''</span> +</span><br><span class="line">                <span class="string">'&#125;'</span>;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">// get set 略</span></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure><p></p><p><strong>改造消息生产者配置</strong></p><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Configuration</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">KafkaProducerConfig</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;spring.kafka.bootstrap-servers&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String bootstrapServers;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Bean</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> ProducerFactory&lt;String, Message&gt; <span class="title">producerFactory</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        Map&lt;String, Object&gt; configProps = <span class="keyword">new</span> HashMap&lt;&gt;();</span><br><span class="line">        configProps.put(</span><br><span class="line">                ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,</span><br><span class="line">                bootstrapServers);</span><br><span class="line">        configProps.put(</span><br><span class="line">                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,</span><br><span class="line">                StringSerializer.class);</span><br><span class="line">        configProps.put(</span><br><span class="line">                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,</span><br><span class="line">                JsonSerializer.class);</span><br><span class="line">        <span class="keyword">return</span> <span class="keyword">new</span> DefaultKafkaProducerFactory&lt;&gt;(configProps);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Bean</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> KafkaTemplate&lt;String, Message&gt; <span class="title">kafkaTemplate</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="keyword">return</span> <span class="keyword">new</span> KafkaTemplate&lt;&gt;(producerFactory());</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure><p></p><p>我们将value序列化策略指定为了Kafka提供的<code>JsonSerializer</code>，并且kafkaTemplate返回类型为<code>KafkaTemplate&lt;String, Message&gt;</code>。</p><p><strong>发送新的消息</strong></p><p>在<code>SendMessageController</code>里发送复杂的消息:</p><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Autowired</span></span><br><span class="line"><span class="keyword">private</span> KafkaTemplate&lt;String, Message&gt; kafkaTemplate;</span><br><span class="line"></span><br><span class="line"><span class="meta">@GetMapping</span>(<span class="string">"send/&#123;message&#125;"</span>)</span><br><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">sendMessage</span><span class="params">(@PathVariable String message)</span> </span>&#123;</span><br><span class="line">    <span class="keyword">this</span>.kafkaTemplate.send(<span class="string">"test"</span>, <span class="keyword">new</span> Message(<span class="string">"mrbird"</span>, message));</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure><p></p><p><strong>修改消费者配置</strong></p><p>修改消费者配置<code>KafkaConsumerConfig</code>：</p><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@EnableKafka</span></span><br><span class="line"><span class="meta">@Configuration</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">KafkaConsumerConfig</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;spring.kafka.bootstrap-servers&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String bootstrapServers;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;spring.kafka.consumer.group-id&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String consumerGroupId;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Value</span>(<span class="string">"$&#123;spring.kafka.consumer.auto-offset-reset&#125;"</span>)</span><br><span class="line">    <span class="keyword">private</span> String autoOffsetReset;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Bean</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> ConsumerFactory&lt;String, Message&gt; <span class="title">consumerFactory</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        Map&lt;String, Object&gt; props = <span class="keyword">new</span> HashMap&lt;&gt;();</span><br><span class="line">        props.put(</span><br><span class="line">                ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,</span><br><span class="line">                bootstrapServers);</span><br><span class="line">        props.put(</span><br><span class="line">                ConsumerConfig.GROUP_ID_CONFIG,</span><br><span class="line">                consumerGroupId);</span><br><span class="line">        props.put(</span><br><span class="line">                ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,</span><br><span class="line">                autoOffsetReset);</span><br><span class="line">        <span class="keyword">return</span> <span class="keyword">new</span> DefaultKafkaConsumerFactory&lt;&gt;(</span><br><span class="line">                props,</span><br><span class="line">                <span class="keyword">new</span> StringDeserializer(),</span><br><span class="line">                <span class="keyword">new</span> JsonDeserializer&lt;&gt;(Message.class));</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Bean</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> ConcurrentKafkaListenerContainerFactory&lt;String, Message&gt; <span class="title">kafkaListenerContainerFactory</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        ConcurrentKafkaListenerContainerFactory&lt;String, Message&gt; factory</span><br><span class="line">                = <span class="keyword">new</span> ConcurrentKafkaListenerContainerFactory&lt;&gt;();</span><br><span class="line">        factory.setConsumerFactory(consumerFactory());</span><br><span class="line">        <span class="keyword">return</span> factory;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure><p></p><p><strong>修改消息监听</strong></p><p>修改<code>KafkaMessageListener</code>：</p><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@KafkaListener</span>(topics = <span class="string">"test"</span>, groupId = <span class="string">"test-consumer"</span>)</span><br><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">listen</span><span class="params">(Message message)</span> </span>&#123;</span><br><span class="line">    logger.info(<span class="string">"接收消息: &#123;&#125;"</span>, message);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure><p></p><p>重启项目，访问<a href="http://localhost:8080/send/hello" target="_blank" rel="noopener">http://localhost:8080/send/hello</a>，控制台输出如下：</p><p><img src="img/QQ截图20190326171125.png" alt="QQ截图20190326171125.png"></p><h3 id="更多配置"><a href="#更多配置" class="headerlink" title="更多配置"></a>更多配置</h3><figure class="highlight yml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br><span class="line">92</span><br><span class="line">93</span><br><span class="line">94</span><br></pre></td><td class="code"><pre><span class="line"><span class="string">spring.kafka.admin.client-id=</span> <span class="comment"># ID to pass to the server when making requests. Used for server-side logging.</span></span><br><span class="line"><span class="string">spring.kafka.admin.fail-fast=false</span> <span class="comment"># Whether to fail fast if the broker is not available on startup.</span></span><br><span class="line"><span class="string">spring.kafka.admin.properties.*=</span> <span class="comment"># Additional admin-specific properties used to configure the client.</span></span><br><span class="line"><span class="string">spring.kafka.admin.ssl.key-password=</span> <span class="comment"># Password of the private key in the key store file.</span></span><br><span class="line"><span class="string">spring.kafka.admin.ssl.key-store-location=</span> <span class="comment"># Location of the key store file.</span></span><br><span class="line"><span class="string">spring.kafka.admin.ssl.key-store-password=</span> <span class="comment"># Store password for the key store file.</span></span><br><span class="line"><span class="string">spring.kafka.admin.ssl.key-store-type=</span> <span class="comment"># Type of the key store.</span></span><br><span class="line"><span class="string">spring.kafka.admin.ssl.protocol=</span> <span class="comment"># SSL protocol to use.</span></span><br><span class="line"><span class="string">spring.kafka.admin.ssl.trust-store-location=</span> <span class="comment"># Location of the trust store file.</span></span><br><span class="line"><span class="string">spring.kafka.admin.ssl.trust-store-password=</span> <span class="comment"># Store password for the trust store file.</span></span><br><span class="line"><span class="string">spring.kafka.admin.ssl.trust-store-type=</span> <span class="comment"># Type of the trust store.</span></span><br><span class="line"><span class="string">spring.kafka.bootstrap-servers=</span> <span class="comment"># Comma-delimited list of host:port pairs to use for establishing the initial connections to the Kafka cluster. Applies to all components unless overridden.</span></span><br><span class="line"><span class="string">spring.kafka.client-id=</span> <span class="comment"># ID to pass to the server when making requests. Used for server-side logging.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.auto-commit-interval=</span> <span class="comment"># Frequency with which the consumer offsets are auto-committed to Kafka if 'enable.auto.commit' is set to true.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.auto-offset-reset=</span> <span class="comment"># What to do when there is no initial offset in Kafka or if the current offset no longer exists on the server.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.bootstrap-servers=</span> <span class="comment"># Comma-delimited list of host:port pairs to use for establishing the initial connections to the Kafka cluster. Overrides the global property, for consumers.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.client-id=</span> <span class="comment"># ID to pass to the server when making requests. Used for server-side logging.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.enable-auto-commit=</span> <span class="comment"># Whether the consumer's offset is periodically committed in the background.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.fetch-max-wait=</span> <span class="comment"># Maximum amount of time the server blocks before answering the fetch request if there isn't sufficient data to immediately satisfy the requirement given by "fetch-min-size".</span></span><br><span class="line"><span class="string">spring.kafka.consumer.fetch-min-size=</span> <span class="comment"># Minimum amount of data the server should return for a fetch request.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.group-id=</span> <span class="comment"># Unique string that identifies the consumer group to which this consumer belongs.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.heartbeat-interval=</span> <span class="comment"># Expected time between heartbeats to the consumer coordinator.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.key-deserializer=</span> <span class="comment"># Deserializer class for keys.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.max-poll-records=</span> <span class="comment"># Maximum number of records returned in a single call to poll().</span></span><br><span class="line"><span class="string">spring.kafka.consumer.properties.*=</span> <span class="comment"># Additional consumer-specific properties used to configure the client.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.ssl.key-password=</span> <span class="comment"># Password of the private key in the key store file.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.ssl.key-store-location=</span> <span class="comment"># Location of the key store file.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.ssl.key-store-password=</span> <span class="comment"># Store password for the key store file.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.ssl.key-store-type=</span> <span class="comment"># Type of the key store.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.ssl.protocol=</span> <span class="comment"># SSL protocol to use.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.ssl.trust-store-location=</span> <span class="comment"># Location of the trust store file.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.ssl.trust-store-password=</span> <span class="comment"># Store password for the trust store file.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.ssl.trust-store-type=</span> <span class="comment"># Type of the trust store.</span></span><br><span class="line"><span class="string">spring.kafka.consumer.value-deserializer=</span> <span class="comment"># Deserializer class for values.</span></span><br><span class="line"><span class="string">spring.kafka.jaas.control-flag=required</span> <span class="comment"># Control flag for login configuration.</span></span><br><span class="line"><span class="string">spring.kafka.jaas.enabled=false</span> <span class="comment"># Whether to enable JAAS configuration.</span></span><br><span class="line"><span class="string">spring.kafka.jaas.login-module=com.sun.security.auth.module.Krb5LoginModule</span> <span class="comment"># Login module.</span></span><br><span class="line"><span class="string">spring.kafka.jaas.options=</span> <span class="comment"># Additional JAAS options.</span></span><br><span class="line"><span class="string">spring.kafka.listener.ack-count=</span> <span class="comment"># Number of records between offset commits when ackMode is "COUNT" or "COUNT_TIME".</span></span><br><span class="line"><span class="string">spring.kafka.listener.ack-mode=</span> <span class="comment"># Listener AckMode. See the spring-kafka documentation.</span></span><br><span class="line"><span class="string">spring.kafka.listener.ack-time=</span> <span class="comment"># Time between offset commits when ackMode is "TIME" or "COUNT_TIME".</span></span><br><span class="line"><span class="string">spring.kafka.listener.client-id=</span> <span class="comment"># Prefix for the listener's consumer client.id property.</span></span><br><span class="line"><span class="string">spring.kafka.listener.concurrency=</span> <span class="comment"># Number of threads to run in the listener containers.</span></span><br><span class="line"><span class="string">spring.kafka.listener.idle-event-interval=</span> <span class="comment"># Time between publishing idle consumer events (no data received).</span></span><br><span class="line"><span class="string">spring.kafka.listener.log-container-config=</span> <span class="comment"># Whether to log the container configuration during initialization (INFO level).</span></span><br><span class="line"><span class="string">spring.kafka.listener.monitor-interval=</span> <span class="comment"># Time between checks for non-responsive consumers. If a duration suffix is not specified, seconds will be used.</span></span><br><span class="line"><span class="string">spring.kafka.listener.no-poll-threshold=</span> <span class="comment"># Multiplier applied to "pollTimeout" to determine if a consumer is non-responsive.</span></span><br><span class="line"><span class="string">spring.kafka.listener.poll-timeout=</span> <span class="comment"># Timeout to use when polling the consumer.</span></span><br><span class="line"><span class="string">spring.kafka.listener.type=single</span> <span class="comment"># Listener type.</span></span><br><span class="line"><span class="string">spring.kafka.producer.acks=</span> <span class="comment"># Number of acknowledgments the producer requires the leader to have received before considering a request complete.</span></span><br><span class="line"><span class="string">spring.kafka.producer.batch-size=</span> <span class="comment"># Default batch size.</span></span><br><span class="line"><span class="string">spring.kafka.producer.bootstrap-servers=</span> <span class="comment"># Comma-delimited list of host:port pairs to use for establishing the initial connections to the Kafka cluster. Overrides the global property, for producers.</span></span><br><span class="line"><span class="string">spring.kafka.producer.buffer-memory=</span> <span class="comment"># Total memory size the producer can use to buffer records waiting to be sent to the server.</span></span><br><span class="line"><span class="string">spring.kafka.producer.client-id=</span> <span class="comment"># ID to pass to the server when making requests. Used for server-side logging.</span></span><br><span class="line"><span class="string">spring.kafka.producer.compression-type=</span> <span class="comment"># Compression type for all data generated by the producer.</span></span><br><span class="line"><span class="string">spring.kafka.producer.key-serializer=</span> <span class="comment"># Serializer class for keys.</span></span><br><span class="line"><span class="string">spring.kafka.producer.properties.*=</span> <span class="comment"># Additional producer-specific properties used to configure the client.</span></span><br><span class="line"><span class="string">spring.kafka.producer.retries=</span> <span class="comment"># When greater than zero, enables retrying of failed sends.</span></span><br><span class="line"><span class="string">spring.kafka.producer.ssl.key-password=</span> <span class="comment"># Password of the private key in the key store file.</span></span><br><span class="line"><span class="string">spring.kafka.producer.ssl.key-store-location=</span> <span class="comment"># Location of the key store file.</span></span><br><span class="line"><span class="string">spring.kafka.producer.ssl.key-store-password=</span> <span class="comment"># Store password for the key store file.</span></span><br><span class="line"><span class="string">spring.kafka.producer.ssl.key-store-type=</span> <span class="comment"># Type of the key store.</span></span><br><span class="line"><span class="string">spring.kafka.producer.ssl.protocol=</span> <span class="comment"># SSL protocol to use.</span></span><br><span class="line"><span class="string">spring.kafka.producer.ssl.trust-store-location=</span> <span class="comment"># Location of the trust store file.</span></span><br><span class="line"><span class="string">spring.kafka.producer.ssl.trust-store-password=</span> <span class="comment"># Store password for the trust store file.</span></span><br><span class="line"><span class="string">spring.kafka.producer.ssl.trust-store-type=</span> <span class="comment"># Type of the trust store.</span></span><br><span class="line"><span class="string">spring.kafka.producer.transaction-id-prefix=</span> <span class="comment"># When non empty, enables transaction support for producer.</span></span><br><span class="line"><span class="string">spring.kafka.producer.value-serializer=</span> <span class="comment"># Serializer class for values.</span></span><br><span class="line"><span class="string">spring.kafka.properties.*=</span> <span class="comment"># Additional properties, common to producers and consumers, used to configure the client.</span></span><br><span class="line"><span class="string">spring.kafka.ssl.key-password=</span> <span class="comment"># Password of the private key in the key store file.</span></span><br><span class="line"><span class="string">spring.kafka.ssl.key-store-location=</span> <span class="comment"># Location of the key store file.</span></span><br><span class="line"><span class="string">spring.kafka.ssl.key-store-password=</span> <span class="comment"># Store password for the key store file.</span></span><br><span class="line"><span class="string">spring.kafka.ssl.key-store-type=</span> <span class="comment"># Type of the key store.</span></span><br><span class="line"><span class="string">spring.kafka.ssl.protocol=</span> <span class="comment"># SSL protocol to use.</span></span><br><span class="line"><span class="string">spring.kafka.ssl.trust-store-location=</span> <span class="comment"># Location of the trust store file.</span></span><br><span class="line"><span class="string">spring.kafka.ssl.trust-store-password=</span> <span class="comment"># Store password for the trust store file.</span></span><br><span class="line"><span class="string">spring.kafka.ssl.trust-store-type=</span> <span class="comment"># Type of the trust store.</span></span><br><span class="line"><span class="string">spring.kafka.streams.application-id=</span> <span class="comment"># Kafka streams application.id property; default spring.application.name.</span></span><br><span class="line"><span class="string">spring.kafka.streams.auto-startup=true</span> <span class="comment"># Whether or not to auto-start the streams factory bean.</span></span><br><span class="line"><span class="string">spring.kafka.streams.bootstrap-servers=</span> <span class="comment"># Comma-delimited list of host:port pairs to use for establishing the initial connections to the Kafka cluster. Overrides the global property, for streams.</span></span><br><span class="line"><span class="string">spring.kafka.streams.cache-max-size-buffering=</span> <span class="comment"># Maximum memory size to be used for buffering across all threads.</span></span><br><span class="line"><span class="string">spring.kafka.streams.client-id=</span> <span class="comment"># ID to pass to the server when making requests. Used for server-side logging.</span></span><br><span class="line"><span class="string">spring.kafka.streams.properties.*=</span> <span class="comment"># Additional Kafka properties used to configure the streams.</span></span><br><span class="line"><span class="string">spring.kafka.streams.replication-factor=</span> <span class="comment"># The replication factor for change log topics and repartition topics created by the stream processing application.</span></span><br><span class="line"><span class="string">spring.kafka.streams.ssl.key-password=</span> <span class="comment"># Password of the private key in the key store file.</span></span><br><span class="line"><span class="string">spring.kafka.streams.ssl.key-store-location=</span> <span class="comment"># Location of the key store file.</span></span><br><span class="line"><span class="string">spring.kafka.streams.ssl.key-store-password=</span> <span class="comment"># Store password for the key store file.</span></span><br><span class="line"><span class="string">spring.kafka.streams.ssl.key-store-type=</span> <span class="comment"># Type of the key store.</span></span><br><span class="line"><span class="string">spring.kafka.streams.ssl.protocol=</span> <span class="comment"># SSL protocol to use.</span></span><br><span class="line"><span class="string">spring.kafka.streams.ssl.trust-store-location=</span> <span class="comment"># Location of the trust store file.</span></span><br><span class="line"><span class="string">spring.kafka.streams.ssl.trust-store-password=</span> <span class="comment"># Store password for the trust store file.</span></span><br><span class="line"><span class="string">spring.kafka.streams.ssl.trust-store-type=</span> <span class="comment"># Type of the trust store.</span></span><br><span class="line"><span class="string">spring.kafka.streams.state-dir=</span> <span class="comment"># Directory location for the state store.</span></span><br><span class="line"><span class="string">spring.kafka.template.default-topic=</span> <span class="comment"># Default topic to which messages are sent.</span></span><br></pre></td></tr></table></figure><p>源码链接：<a href="https://github.com/wuyouzhuguli/SpringAll/tree/master/54.Spring-Boot-Kafka" target="_blank" rel="noopener">https://github.com/wuyouzhuguli/SpringAll/tree/master/54.Spring-Boot-Kafka</a></p><script>$(".post-body a").not(".thispage").addClass("ignore-href").attr("target","_blank")</script></div><div></div><div><div style="padding:10px 0;margin:20px auto;width:90%;text-align:center;color:#878787"><div>请作者喝瓶肥宅水~</div><button id="rewardButton" style="margin-top:10px" disable="enable" onclick='var e=document.getElementById("QR");"none"===e.style.display?e.style.display="block":e.style.display="none"'><span style="height:46px;width:46px;line-height:46px;border-radius:50%;color:#fe5f55;font-weight:600;background:#ffd5be;border:none;box-shadow:0 4px 8px 0 rgba(255,213,190,.4)">￥</span></button><div id="QR" style="display:none"><div id="wechat" style="display:inline-block"><img id="wechat_qr" src="/img/wechat_pay.png" alt="MrBird WeChat Pay"></div><div id="alipay" style="display:inline-block"><img id="alipay_qr" src="/img/ali_pay.png" alt="MrBird Alipay"></div></div></div><style>#QR img{width:auto;margin:.8em 1em 0 1em}</style></div><div><ul class="post-copyright"><li class="post-copyright-author"><strong>本文作者：</strong> MrBird</li><li class="post-copyright-link"><strong>本文链接：</strong> <a href="http://mrbird.cc/Spring-Boot-Kafka.html" title="Spring Boot整合Kafka">http://mrbird.cc/Spring-Boot-Kafka.html</a></li><li class="post-copyright-license"><strong>版权声明： </strong>本博客所有文章除特别声明外，均采用 <a href="https://creativecommons.org/licenses/by-nc-sa/4.0/" rel="external nofollow" target="_blank">CC BY-NC-SA 4.0</a> 许可协议。转载请注明出处！</li></ul></div><footer class="post-footer"><div class="post-tags" style="margin-bottom:1.3rem"><a href="/tags/Spring-Boot/" rel="tag"># Spring Boot</a> <a href="/tags/Kafka/" rel="tag"># Kafka</a></div><div class="post-nav"><div class="post-nav-next post-nav-item"><a href="/Spring-Boot-2-0-WebFlux.html" rel="next" title="Spring Boot 2.0 WebFlux编程"><i class="fa fa-chevron-left"></i> Spring Boot 2.0 WebFlux编程</a></div><span class="post-nav-divider"></span><div class="post-nav-prev post-nav-item"><a href="/Spring-Cloud-Consul.html" rel="prev" title="Spring Cloud Consul服务治理">Spring Cloud Consul服务治理 <i class="fa fa-chevron-right"></i></a></div></div></footer></article><hr><div id="container"></div><div class="post-spread"><div id="comment-div"></div><style>.valine .vlist{margin-bottom:3rem}.valine .vwrap .vcontrol .col.col-60{text-align:left}.valine .vlist .vcard .vhead,.valine .vlist .vcard section .vfooter{text-align:left}.valine .vlist .vcard section{padding-bottom:.5rem!important}.vname{color:#42b983!important}.valine .vinfo .col{text-align:left;margin-left:-27rem}div#comment-div{margin-bottom:-8rem}.valine .vlist .vcard .vcontent .code,.valine .vlist .vcard .vcontent code,.valine .vlist .vcard .vcontent pre{background:#fbfbfb}.valine .vlist .vcard .vcontent a{color:#42b983}.valine .vlist .vcard .vimg{border-radius:3px}.valine .vbtn{border-radius:2px;padding:.3rem 1.25rem}.valine .vbtn:active,.valine .vbtn:hover{color:#42b983;border-color:#42b983;background-color:#fff}.valine .vwrap .vheader .vinput:focus{border-bottom-color:#42b983}.valine .vlist .vcard .vcontent.expand:before{background:-webkit-gradient(linear,left top,left bottom,from(hsla(0,0%,100%,0)),to(hsla(0,0%,100%,.2)));background:linear-gradient(180deg,hsla(0,0%,100%,0),hsla(0,0%,100%,.2))}.valine .vlist .vcard .vcontent.expand:after{content:"点击展开";font-size:.4rem;text-align:right;left:-1rem;background:hsla(0,0%,100%,.2)}.valine .vlist .vcard section .vfooter .vat{color:#b3b3b3}.valine .vlist .vcard section .vfooter .vat:hover{color:#42b983}.vcontent img{margin:0}.valine .info{display:none}</style><script type="text/javascript" src="/js/av.min.js"></script><script type="text/javascript" src="/js/Valine.min.js"></script><script type="text/javascript" src="/js/activate-power-mode.js"></script><script>POWERMODE.colorful=!0,POWERMODE.shake=!1,document.body.addEventListener("input",POWERMODE),new Valine({el:"#comment-div",notify:!1,verify:!0,appId:"SMcDFP1bN1jgb9Lo8JmtICHm-gzGzoHsz",appKey:"dH4nrUrt3V5XiJiI6Qyejnbi",placeholder:"",path:window.location.pathname,avatar:"monsterid",guest_info:["nick","mail","link"]});var chicken='<a href="#"><img src="https://image.uisdc.com/wp-content/uploads/2018/06/uisdc-chat-chicken.gif" class="checken"></a>';$("#comment-div").prepend(chicken)</script></div></div><script>var $bqinline=$("img[alt='bq-inline']");$bqinline.css({width:"2.3rem",height:"2.3rem",display:"inline","vertical-align":"text-bottom"})</script></div><div class="comments" id="comments"></div></div><aside id="sidebar" class="sidebar" onselectstart="return!1"><div class="sidebar-inner"><ul class="sidebar-nav motion-element"><li class="sidebar-nav-toc sidebar-nav-active" data-target="post-toc-wrap">Contents</li><li class="sidebar-nav-overview" data-target="site-overview">Site Preview</li></ul><section class="site-overview sidebar-panel"><div class="sidebar-sticky sticky"><div itemscope itemtype="http://schema.org/Person"><div class="author__avatar"><img src="/images/blogImage.jpg" class="author__avatar" alt="MrBird" itemprop="image"></div><div class="author__content"><h3 class="author__name" itemprop="name">MrBird's Blog</h3><p class="author__bio" itemprop="description">A simple blog, code repository, just keep blogging</p></div><div class="author__urls-wrapper"><button class="btn btn--inverse">Follow</button><ul class="author__urls social-icons"><li><a href="http://map.baidu.com/?newmap=1&s=s%26wd%3D%E7%A6%8F%E5%B7%9E%E5%B8%82%26c%3D300&from=alamap&tpl=mapcity" target="_blank" itemprop="url" class="ignore-href"><i class="fa fa-fw fa-map-marker" aria-hidden="true"></i>&nbsp;&nbsp;FuZhou,CN</a></li><li><a href="https://love.mrbird.cc" target="_blank" itemprop="url" class="ignore-href"><i class="fa fa-fw fa-diamond" aria-hidden="true"></i>&nbsp;&nbsp;Love</a></li><li><a href="https://cloud.mrbird.cn" target="_blank" itemprop="url" class="ignore-href"><i class="fa fa-fw fa-chain" aria-hidden="true"></i>&nbsp;&nbsp;FEBS</a></li><li><a href="/atom.xml" target="_blank" itemprop="url" class="ignore-href"><i class="fa fa-fw fa-rss" aria-hidden="true"></i>&nbsp;&nbsp;RSS</a></li><li><a href="https://gitee.com/mrbirdd" target="_blank" itemprop="sameAs" class="ignore-href"><i class="fa fa-fw fa-codepen" aria-hidden="true"></i>&nbsp;&nbsp;Gitee</a></li><li><a href="https://github.com/wuyouzhuguli" target="_blank" itemprop="sameAs" class="ignore-href"><i class="fa fa-fw fa-github-alt" aria-hidden="true"></i>&nbsp;&nbsp;GitHub</a></li><li><a href="javascript:;" class="popup-trigger"><i class="fa fa-fw fa-search" aria-hidden="true"></i>&nbsp;&nbsp;Search</a></li></ul></div></div></div><script>var $urls=$(".author__urls").find("a");$urls.each(function(){var o=$(this);o.mouseenter(function(){o.css({color:"#414547"})}),o.mouseleave(function(){o.css({color:""})})})</script></section><section class="post-toc-wrap motion-element sidebar-panel sidebar-panel-active"><div class="post-toc"><div class="post-toc-content"><ol class="nav"><li class="nav-item nav-level-2"><a class="nav-link" href="#Kafka安装使用"><span class="nav-number">1.</span> <span class="nav-text">Kafka安装使用</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#启动Zookeeper"><span class="nav-number">1.1.</span> <span class="nav-text">启动Zookeeper</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#启动Kafka"><span class="nav-number">1.2.</span> <span class="nav-text">启动Kafka</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#创建Topic"><span class="nav-number">1.3.</span> <span class="nav-text">创建Topic</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#生产消息和消费消息"><span class="nav-number">1.4.</span> <span class="nav-text">生产消息和消费消息</span></a></li></ol></li><li class="nav-item nav-level-2"><a class="nav-link" href="#Spring-Boot整合Kafaka"><span class="nav-number">2.</span> <span class="nav-text">Spring Boot整合Kafaka</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#生产者配置"><span class="nav-number">2.1.</span> <span class="nav-text">生产者配置</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#发布消息"><span class="nav-number">2.2.</span> <span class="nav-text">发布消息</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#消费者配置"><span class="nav-number">2.3.</span> <span class="nav-text">消费者配置</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#消息消费"><span class="nav-number">2.4.</span> <span class="nav-text">消息消费</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#演示"><span class="nav-number">2.5.</span> <span class="nav-text">演示</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#KafkaListener详解"><span class="nav-number">2.6.</span> <span class="nav-text">@KafkaListener详解</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#消息过滤器"><span class="nav-number">2.7.</span> <span class="nav-text">消息过滤器</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#发送复杂的消息"><span class="nav-number">2.8.</span> <span class="nav-text">发送复杂的消息</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#更多配置"><span class="nav-number">2.9.</span> <span class="nav-text">更多配置</span></a></li></ol></li></ol></div></div></section></div></aside></div></main><footer id="footer" class="footer" onselectstart="return!1"><div class="footer-inner"><div class="copyright">&copy; 2016 - <span itemprop="copyrightYear">2019</span>&nbsp;&nbsp; <span class="author" itemprop="copyrightHolder">MrBird</span>&nbsp;&nbsp;|<script async src="//busuanzi.ibruce.info/busuanzi/2.3/busuanzi.pure.mini.js"></script>&nbsp;&nbsp;UV&nbsp;<span class="busuanzi-value" id="busuanzi_value_site_uv" style="cursor:pointer" title="统计开始时间：2019年7月5日"></span> &nbsp;&nbsp;PV&nbsp;<span class="busuanzi-value" id="busuanzi_value_site_pv" style="cursor:pointer" title="统计开始时间：2019年7月5日"></span></div></div></footer><div class="back-to-top"><span style="font-family:'Source Sans Pro','Helvetica Neue',Arial,sans-serif;font-size:1.2em;font-weight:600">TOP</span></div></div><script type="text/javascript">"[object Function]"!==Object.prototype.toString.call(window.Promise)&&(window.Promise=null)</script><script type="text/javascript" src="/lib/jquery/index.js?v=2.1.3"></script><script type="text/javascript" src="/lib/fastclick/lib/fastclick.min.js?v=1.0.6"></script><script type="text/javascript" src="/lib/jquery_lazyload/jquery.lazyload.js?v=1.9.7"></script><script type="text/javascript" src="/lib/velocity/velocity.min.js?v=1.2.1"></script><script type="text/javascript" src="/lib/velocity/velocity.ui.min.js?v=1.2.1"></script><script type="text/javascript" src="/js/src/utils.js?v=5.1.1"></script><script type="text/javascript" src="/js/src/motion.js?v=5.1.1"></script><script type="text/javascript" src="/js/src/scrollspy.js?v=5.1.1"></script><script type="text/javascript" src="/js/src/post-details.js?v=5.1.1"></script><script type="text/javascript" src="/js/src/bootstrap.js?v=5.1.1"></script><script type="text/javascript">function proceedsearch(){$("body").append('<div class="search-popup-overlay local-search-pop-overlay"></div>').css("overflow","hidden"),$(".search-popup-overlay").click(onPopupClose),$(".popup").toggle();var t=$("#local-search-input");t.attr("autocapitalize","none"),t.attr("autocorrect","off"),t.focus()}var isfetched=!1,isXml=!0,search_path="search.xml";0===search_path.length?search_path="search.xml":search_path.endsWith("json")&&(isXml=!1);var path="/"+search_path,onPopupClose=function(t){$(".popup").hide(),$("#local-search-input").val(""),$(".search-result-list").remove(),$("#no-result").remove(),$(".local-search-pop-overlay").remove(),$("body").css("overflow","")},searchFunc=function(t,e,o){"use strict";$("body").append('<div class="search-popup-overlay local-search-pop-overlay"><div id="search-loading-icon"><i class="fa fa-spinner fa-pulse fa-2x fa-fw"></i></div></div>').css("overflow","hidden"),$("#search-loading-icon").css("margin","20% auto 0 auto").css("text-align","center"),$.ajax({url:t,dataType:isXml?"xml":"json",async:!0,success:function(t){isfetched=!0,$(".popup").detach().appendTo(".header-inner");var n=isXml?$("entry",t).map(function(){return{title:$("title",this).text(),content:$("content",this).text(),url:$("url",this).text()}}).get():t,r=document.getElementById(e),s=document.getElementById(o),a=function(){var t=r.value.trim().toLowerCase(),e=t.split(/[\s\-]+/);e.length>1&&e.push(t);var o=[];if(t.length>0&&n.forEach(function(n){function r(e,o,n,r){for(var s=r[r.length-1],a=s.position,i=s.word,l=[],h=0;a+i.length<=n&&0!=r.length;){i===t&&h++,l.push({position:a,length:i.length});var p=a+i.length;for(r.pop();0!=r.length&&(s=r[r.length-1],a=s.position,i=s.word,p>a);)r.pop()}return c+=h,{hits:l,start:o,end:n,searchTextCount:h}}function s(t,e){var o="",n=e.start;return e.hits.forEach(function(e){o+=t.substring(n,e.position);var r=e.position+e.length;o+='<b class="search-keyword">'+t.substring(e.position,r)+"</b>",n=r}),o+=t.substring(n,e.end)}var a=!1,i=0,c=0,l=n.title.trim(),h=l.toLowerCase(),p=n.content.trim().replace(/<[^>]+>/g,""),u=p.toLowerCase(),f=decodeURIComponent(n.url),d=[],g=[];if(""!=l&&(e.forEach(function(t){function e(t,e,o){var n=t.length;if(0===n)return[];var r=0,s=[],a=[];for(o||(e=e.toLowerCase(),t=t.toLowerCase());(s=e.indexOf(t,r))>-1;)a.push({position:s,word:t}),r=s+n;return a}d=d.concat(e(t,h,!1)),g=g.concat(e(t,u,!1))}),(d.length>0||g.length>0)&&(a=!0,i=d.length+g.length)),a){[d,g].forEach(function(t){t.sort(function(t,e){return e.position!==t.position?e.position-t.position:t.word.length-e.word.length})});var v=[];0!=d.length&&v.push(r(l,0,l.length,d));for(var C=[];0!=g.length;){var $=g[g.length-1],m=$.position,x=$.word,w=m-20,y=m+80;w<0&&(w=0),y<m+x.length&&(y=m+x.length),y>p.length&&(y=p.length),C.push(r(p,w,y,g))}C.sort(function(t,e){return t.searchTextCount!==e.searchTextCount?e.searchTextCount-t.searchTextCount:t.hits.length!==e.hits.length?e.hits.length-t.hits.length:t.start-e.start});var T=parseInt("1");T>=0&&(C=C.slice(0,T));var b="";b+=0!=v.length?"<li><a href='"+f+"' class='search-result-title'>"+s(l,v[0])+"</a>":"<li><a href='"+f+"' class='search-result-title'>"+l+"</a>",C.forEach(function(t){b+="<a href='"+f+'\'><p class="search-result">'+s(p,t)+"...</p></a>"}),b+="</li>",o.push({item:b,searchTextCount:c,hitCount:i,id:o.length})}}),1===e.length&&""===e[0])s.innerHTML='<div id="no-result"><i class="fa fa-search fa-5x" /></div>';else if(0===o.length)s.innerHTML='<div id="no-result"><i class="fa fa-frown-o fa-5x" /></div>';else{o.sort(function(t,e){return t.searchTextCount!==e.searchTextCount?e.searchTextCount-t.searchTextCount:t.hitCount!==e.hitCount?e.hitCount-t.hitCount:e.id-t.id});var a='<ul class="search-result-list">';o.forEach(function(t){a+=t.item}),a+="</ul>",s.innerHTML=a}};r.addEventListener("input",a),$(".local-search-pop-overlay").remove(),$("body").css("overflow",""),proceedsearch()}})};$(".popup-trigger").click(function(t){t.stopPropagation(),isfetched===!1?searchFunc(path,"local-search-input","local-search-result"):proceedsearch()}),$(".popup-btn-close").click(onPopupClose),$(".popup").click(function(t){t.stopPropagation()}),$(document).on("keyup",function(t){var e=27===t.which&&$(".search-popup").is(":visible");e&&onPopupClose()})</script></body><script>$(function(){$("a").not(".nav-link,.cloud-tie-join-count,.ignore-href,.jstree-anchor").addClass("animsition-link")});var burst1=new mojs.Burst({left:0,top:0,radius:{5:40},children:{shape:"circle",fill:["red","cyan","orange"],fillOpacity:.8,radiusX:3.5,radiusY:3.5}});document.addEventListener("click",function(a){null==a.target.href&&"footer"!=a.target.className&&"copyright"!=a.target.className&&"author__urls social-icons"!=a.target.className&&"author__avatar"!=a.target.className&&"sidebar sidebar-active"!=a.target.className&&burst1.tune({x:a.pageX,y:a.pageY}).generate().replay()})</script><script type="text/javascript" src="/js/message.js"></script></html><!-- rebuild by neat -->